package test
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
object ll {
      def main(args: Array[String]): Unit = {
        val spark = SparkSession.builder()
          .appName("TrafficFlowStatistics")
          .master("local[*]")
          .getOrCreate()

        val trafficData = spark.read.text("traffic-data.txt")

        val provinceFlow = trafficData
          .select(substring($"value", 1, 1).alias("province"))
          .groupBy("province")
          .count()
          .withColumnRenamed("count", "flowCount")

        val jdbcUrl = "jdbc:mysql://localhost:3306/traffic"
        val connectionProperties = new java.util.Properties()
        connectionProperties.put("user", "root")
        connectionProperties.put("password", "123456")
        connectionProperties.put("driver", "com.mysql.cj.jdbc.Driver") // 明确指定驱动类

        provinceFlow.write
          .jdbc(jdbcUrl, "traffic", connectionProperties)

        val result = spark.read
          .jdbc(jdbcUrl, "traffic", connectionProperties)
        result.show()

        spark.stop()
      }
    }

